package rts

import (
	"context"
	"time"

	"github.com/spf13/cast"

	"code.byted.org/bytertc/whiteboard_server/biz/dal/ddl"
	frontierSDK "code.byted.org/frontier/serverSDK/go"

	"code.byted.org/bytertc/whiteboard_server/biz/api"

	bytedtracer "code.byted.org/bytedtrace/interface-go"
	v1 "code.byted.org/bytertc/whiteboard_server/biz/api/pb/v1"
	"code.byted.org/bytertc/whiteboard_server/biz/config"
	"code.byted.org/bytertc/whiteboard_server/biz/dal"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/biz_error"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/gateway"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/trace"
	"code.byted.org/bytertc/whiteboard_server/biz/service/command"
	"code.byted.org/bytertc/whiteboard_server/biz/service/command/factory"
	"code.byted.org/gopkg/gopool"
	"code.byted.org/gopkg/logs/v2/log"
	"code.byted.org/videoarch/universal_utils_go/event_collector/tracer"
	"code.byted.org/videoarch/universal_utils_go/event_collector/util"
	"github.com/pkg/errors"
	"google.golang.org/protobuf/proto"
)

const (
	SourceOpenapi = "source_openapi"
	SourceWs      = "source_ws"
	SourceRtsSDK  = "source_rts_sdk"
)

const (
	WsPayloadType     = "text/plain"
	WsPayloadEncoding = ""
)

type Ext struct {
	Source string
	WsExt  WsExt
}

type WsExt struct {
	ProductID       int32
	AppID           int32
	DeviceID        int64
	UserID          int64
	Service         int32
	Method          int32
	PayloadEncoding string
	PayloadType     string
}

type Adapter interface {
	HandleMessage(ctx context.Context, params *api.AuthTokenKeyParams, message string, ext Ext) error
}

type adapter struct {
	Repo    dal.Repo
	cfg     config.DynamicConfig
	rtcAPI  gateway.RtcAPI
	manager Manager
}

func NewAdapter(repo dal.Repo, api gateway.RtcAPI, dynamicConfig config.DynamicConfig, manager Manager) *adapter {
	return &adapter{
		Repo:    repo,
		rtcAPI:  api,
		cfg:     dynamicConfig,
		manager: manager,
	}
}

func (a *adapter) HandleMessage(ctx context.Context, params *api.AuthTokenKeyParams, message string, ext Ext) error {
	start := time.Now()
	request, err := decodeRTM(message)
	if err != nil {
		return err
	}
	bytedtracer.AddInfoLogEvent(ctx, trace.HandleRTM.String(), "RTM Message", bytedtracer.NewTagKV("traceID", request.GetTraceId()), bytedtracer.NewTagKV("AppID", request.AppId), bytedtracer.NewTagKV("roomID", request.RoomId), bytedtracer.NewTagKV("userID", request.Operator), bytedtracer.NewTagKV("command", request.Command), bytedtracer.NewTagKV("msg", message))
	if ext.Source == SourceOpenapi {
		if params.AppID != request.AppId {
			return biz_error.InvalidParam.Wrap(errors.Errorf("params invaild token appid:%s,msg appid:%s", params.AppID, request.AppId))
		}
		if params.RoomID != "" && request.RoomId != "" && params.RoomID != request.RoomId {
			return biz_error.InvalidParam.Wrap(errors.Errorf("params invaild token roomid:%s,msg roomid:%s", params.RoomID, request.RoomId))
		}
		if params.Operator != request.Operator {
			return biz_error.InvalidParam.Wrap(errors.Errorf("params invaild token operator:%s,msg operator:%s", params.Operator, request.Operator))
		}
	}
	if ext.Source == SourceRtsSDK {
		a.manager.UpdateLastActiveTime(ctx, request.AppId, request.RoomId)
	}

	span, cmdCtx := bytedtracer.StartCustomSpan(ctx, trace.INTERNAL, trace.CommandTraceName(request.Command),
		bytedtracer.EnableEmitSpanMetrics)
	defer span.Finish()
	span.SetTags(bytedtracer.NewTagKV("app_id", request.AppId),
		bytedtracer.NewTagKV("room_id", request.RoomId),
		bytedtracer.NewTagKV("operator", request.Operator),
		bytedtracer.NewTagKV("trace_id", request.GetTraceId()))

	cmd, err := factory.New(cmdCtx, a.Repo, a.cfg, request)
	if err != nil {
		if name, ok := v1.Command_name[int32(request.Command)]; ok {
			traceKey := &util.KeyWord{
				RoomId: request.RoomId,
				AppId:  request.AppId,
				Tag:    name,
				UserId: request.Operator,
				Failed: 1,
			}
			tracer.MileStone(tracer.STREAM_COLLECTOR, "send_rtm_fail", "", traceKey, nil)
		}
		return err
	}
	results := cmd.Execute(cmdCtx)
	for _, result := range results {
		if result.Err() != nil {
			log.V1.CtxError(cmdCtx, "%+v", result.Err())
			trace.TagServerSpanError(ctx, result.Err())
		}
		a.handleResult(ctx, request, result, ext)
	}
	traceKey := &util.KeyWord{
		RoomId: request.RoomId,
		AppId:  request.AppId,
		UserId: request.Operator,
	}
	otherInfo := map[string]string{
		"command":   request.Command.String(),
		"action":    "",
		"elapse":    cast.ToString(time.Since(start).Milliseconds()),
		"timestamp": cast.ToString(start.UnixMilli()),
	}

	go tracer.MileStone(tracer.STREAM_COLLECTOR, "wb_process_elapse", "", traceKey, otherInfo)

	return nil
}

func (a *adapter) handleResult(ctx context.Context, request *v1.OnWireRequest, result command.Result, ext Ext) {
	data, err := GenMsgFromResult(ctx, request, result)
	traceKey := &util.KeyWord{
		RoomId: request.RoomId,
		AppId:  request.AppId,
		UserId: request.Operator,
		Tag:    "unknown",
	}
	if name, ok := v1.Command_name[int32(request.Command)]; ok {
		traceKey.Tag = name
	}
	traceFilter := false
	if _, ok := traceFilterCommand[traceKey.Tag]; ok {
		traceFilter = true
	}

	if err != nil {
		log.V1.CtxError(ctx, "fail to handle result %+v", err)
		return
	}

	wsParam, err := a.Repo.GetWsParam(ctx, request.AppId, request.RoomId)
	if err != nil {
		log.V1.CtxError(ctx, "GetWsParam error:%+v", err)
	}
	if result.Broadcast() {
		a.toRoom(ctx, request, data, wsParam, traceKey, traceFilter, ext)
	} else {
		a.toUser(ctx, request, data, traceKey, traceFilter, ext)
	}

}

func (a *adapter) toRoom(ctx context.Context, request *v1.OnWireRequest, data string, wsParam *ddl.WsParam, traceKey *util.KeyWord, traceFilter bool, ext Ext) {
	if ext.Source == SourceRtsSDK {
		gopool.CtxGo(ctx, func() {
			log.V1.CtxInfo(ctx, "send rts sdk data:%v", data)
			err := a.manager.SendBroadcastMsg(ctx, request.AppId, request.RoomId, data)
			if err != nil {
				err = a.manager.SendBroadcastMsg(ctx, request.AppId, request.RoomId, data)
				if err != nil {
					log.V1.CtxError(ctx, "rts proxy send broadcast msg error:%v,AppID:%s,roomID:%s", err, request.AppId, request.RoomId)
					return
				}
			}
		})
		return
	}
	data = EncodeBase64([]byte(data))

	if wsParam != nil {
		wsReq := frontierSDK.NewGroupPushRequest()
		wsReq.ProductID = wsParam.WsProductID
		wsReq.AppID = wsParam.WsAppID
		wsReq.Service = wsParam.WsService
		wsReq.Group = wsParam.WsGroupID
		wsReq.PayloadEncoding = WsPayloadEncoding
		wsReq.PayloadType = WsPayloadType
		wsReq.Payload = []byte(data)
		wsResp, err := frontierSDK.GroupPush(ctx, wsReq)
		log.V1.CtxInfo(ctx, "ws req:%+v,resp:%+v", wsReq, wsResp)
		if err != nil {
			log.V1.CtxError(ctx, "frontier push error:%+v,req:%+v,resp", err, wsReq, wsResp)
		}
	}

	gopool.CtxGo(ctx, func() {
		err, errCode := a.rtcAPI.ToRoom(ctx, request.AppId, request.RoomId, data, traceKey.Tag)
		if !traceFilter {
			if err != nil {
				traceKey.Failed = 1
				tracer.MileStone(tracer.STREAM_COLLECTOR, "send_rtm_fail", "handleResult.ToRoom", traceKey, map[string]string{"err_code": errCode})
			} else {
				tracer.MileStone(tracer.STREAM_COLLECTOR, "send_rtm_success", "handleResult.ToRoom", traceKey, map[string]string{"err_code": errCode})
			}
		}

	})
}

func (a *adapter) toUser(ctx context.Context, request *v1.OnWireRequest, data string, traceKey *util.KeyWord, traceFilter bool, ext Ext) {
	if ext.Source == SourceRtsSDK {
		gopool.CtxGo(ctx, func() {
			log.V1.CtxInfo(ctx, "send rts sdk data:%v", data)
			err := a.manager.SendUnicastMsg(ctx, request.AppId, request.Operator, data)
			if err != nil {
				err = a.manager.SendUnicastMsg(ctx, request.AppId, request.Operator, data)
				if err != nil {
					log.V1.CtxError(ctx, "rts proxy send unicast msg error:%v,AppID:%s,userID:%s", err, request.AppId, request.Operator)
					return
				}
			}
		})
		return
	}
	data = EncodeBase64([]byte(data))

	if ext.Source == SourceWs {
		wsReq := frontierSDK.NewPushRequest()
		wsReq.ProductID = ext.WsExt.ProductID
		wsReq.AppID = ext.WsExt.AppID
		wsReq.Service = ext.WsExt.Service
		wsReq.Method = ext.WsExt.Method
		wsReq.Devices = []*frontierSDK.Device{{UserID: ext.WsExt.UserID, DeviceID: ext.WsExt.DeviceID}}
		wsReq.PayloadEncoding = WsPayloadEncoding
		wsReq.PayloadType = WsPayloadType
		wsReq.Payload = []byte(data)

		wsResp, err := frontierSDK.Push(ctx, wsReq)
		log.V1.CtxInfo(ctx, "ws req:%+v,resp:%+v", wsReq, wsResp)
		if err != nil {
			log.V1.CtxError(ctx, "frontier push error:%+v,req:%+v,resp", err, wsReq, wsResp)
		}
		return
	}
	gopool.CtxGo(ctx, func() {
		err, errCode := a.rtcAPI.ToUser(ctx, request.AppId, request.RoomId, request.Operator, data, traceKey.Tag, request.RtsP2PType)
		if !traceFilter {
			if err != nil {
				traceKey.Failed = 1
				tracer.MileStone(tracer.STREAM_COLLECTOR, "send_rtm_fail", "handleResult.ToUser", traceKey, map[string]string{"err_code": errCode})
			} else {
				tracer.MileStone(tracer.STREAM_COLLECTOR, "send_rtm_success", "handleResult.ToUser", traceKey, map[string]string{"err_code": errCode})
			}
		}

	})
}

func GenMsgFromResult(ctx context.Context, request *v1.OnWireRequest, result command.Result) (string, error) {
	data, err := proto.Marshal(result.Message())
	if err != nil {
		return "", biz_error.EncodeError.Wrap(errors.WithStack(err))
	}
	resp := &v1.OnWireResponse{
		Operator: request.Operator,
		Command:  result.Command(),
		Data:     data,
		RoomId:   &(request.RoomId),
		TraceId:  request.TraceId,
	}
	if result.Err() != nil {
		errMsg := result.Err().Error()
		if len(errMsg) > 100 {
			errMsg = "the error is too long to provide"
		}
		resp.ErrMsg = &errMsg
	}
	data, err = proto.Marshal(resp)
	return string(data), biz_error.EncodeError.Wrap(errors.WithStack(err))
}

// todo 发送：截断超长的 packet。应该只有 snapshot 会有可能超过 64 KB，editing 的回复长度不会超过请求的长度

var traceFilterCommand = map[string]bool{
	"kGetBoardInfo":     true,
	"kGetActionCount":   true,
	"kGetShapeSnapshot": true,
	"kHeartBeat":        true,
}
